package cn.azzhu.day02.source;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.NumberSequenceIterator;

import java.util.Arrays;

/**
 * 可以并行的的source
 * @author azzhu
 * @create 2020-09-17 23:36:36
 */
public class SourceDemo2 {
    public static void main(String[] args) throws Exception {
        //实时计算环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //final DataStreamSource<Long> nums = env.fromParallelCollection(new NumberSequenceIterator(1, 10), Long.class);
        //final DataStreamSource<Long> nums = env.fromParallelCollection(new NumberSequenceIterator(1, 10), TypeInformation.of(Long.class));

        final DataStreamSource<Long> nums = env.generateSequence(1, 100);

        final int parallelism = nums.getParallelism();
        System.out.println("==============>"+parallelism);  //todo 8

        final SingleOutputStreamOperator<Long> filtered = nums.filter(num -> num % 2 == 0).setParallelism(3);

        final int parallelism1 = filtered.getParallelism();
        System.out.println("&&&&&&&&&&& >>>>" + parallelism1);  //todo 3
        filtered.print();

        env.execute("SourceDemo1");
    }
}
